package com.niit.kafka;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;
import java.util.Properties;

public class KafkaApplication {

    public static void main(String[] args) {
        String brokers = "hadoop001:9092"; // 定义kfaka服务连接地址
        String zookeepers = "hadoop001:2181";  // 定义zookeeper连接地址
        String from = "log"; // 定义名字为log的输入topic
        String to = "recommender"; // 定义名字为输出的topic

        // 定义kafka streaming的配置
        Properties settings = new Properties();
        settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter");
        settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, zookeepers);
        settings.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);

        // 创建StreamsConfig对象，用于配置KafkaStreaming实例
        StreamsConfig streamsConfig = new StreamsConfig(settings);
        // 拓扑建构器
        TopologyBuilder builder = new TopologyBuilder();
        builder.addSource("SOURCE",from)
                .addProcessor("PROCESS",() -> new LogProcessor(), "SOURCE")
                .addSink("SINK",to,"PROCESS");
        // 创建kafkaStreaming对象
        KafkaStreams kafkaStreams = new KafkaStreams(builder,streamsConfig);
        // 启动KafkaStreaming
        kafkaStreams.start();
        System.out.println("kafka stream started!");
    }
}
